package com.lry;

import com.lry.config.BrokerRole;
import com.lry.config.MessageStoreConfig;
import com.lry.config.StorePathConfigHelper;
import com.lry.index.IndexService;
import com.lry.message.MessageDecoder;
import com.lry.message.MessageExt;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;

import java.io.File;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.ByteBuffer;
import java.nio.channels.FileLock;
import java.util.LinkedList;
import java.util.concurrent.*;

@Data
@Slf4j
public class DefaultMessageStore {

    private SystemClock systemClock = new SystemClock();

    private final MessageStoreConfig messageStoreConfig;

    private final ConcurrentMap<String/* topic */, ConcurrentMap<Integer/* queueId */, ConsumeQueue>> consumeQueueTable;

    private final FlushConsumeQueueService flushConsumeQueueService;

    private final CommitLog commitLog;

    private final IndexService indexService;

    private final ReputMessageService reputMessageService;

    private final ScheduledExecutorService scheduledExecutorService =
            Executors.newSingleThreadScheduledExecutor(new ThreadFactoryImpl("StoreScheduledThread"));

    private final LinkedList<CommitLogDispatcher> dispatcherList;

    private RandomAccessFile lockFile;

    private FileLock lock;

    private volatile boolean shutdown;

    public DefaultMessageStore(MessageStoreConfig messageStoreConfig) throws IOException {
        this.messageStoreConfig = messageStoreConfig;
        this.flushConsumeQueueService = new FlushConsumeQueueService();
        this.consumeQueueTable = new ConcurrentHashMap<>();
        this.commitLog = new CommitLog(this);
        this.dispatcherList = new LinkedList<>();
        this.dispatcherList.addLast(new CommitLogDispatcherBuildConsumeQueue());
        this.dispatcherList.addLast(new CommitLogDispatcherBuildIndex());
        this.indexService = new IndexService(this);

        File file = new File(StorePathConfigHelper.getLockFile(messageStoreConfig.getStorePathRootDir()));
        MappedFile.ensureDirOK(file.getParent());
        MappedFile.ensureDirOK(getStorePathPhysic());
        MappedFile.ensureDirOK(getStorePathLogic());
        lockFile = new RandomAccessFile(file, "rw");
        this.load();

        this.reputMessageService = new ReputMessageService(this.commitLog.getReputFromOffset());
    }


    public void start() throws Exception {

        lock = lockFile.getChannel().tryLock(0, 1, false);
        if (lock == null || lock.isShared() || !lock.isValid()) {
            throw new RuntimeException("Lock failed,MQ already started");
        }

        lockFile.getChannel().write(ByteBuffer.wrap("lock".getBytes()));
        lockFile.getChannel().force(true);
        this.reputMessageService.start();
        this.flushConsumeQueueService.start();
        this.commitLog.start();

        this.shutdown = false;
    }

    public PutMessageResult putMessage(MessageExt msg) {
        return waitForPutResult(asyncPutMessage(msg));
    }


    public boolean load() {
        boolean result = true;

        try {
            // load Commit Log
            result = result && this.commitLog.load();

            // load Consume Queue
            result = result && this.loadConsumeQueue();

            if (result) {
                this.indexService.load();
            }

        } catch (Exception e) {
            log.error("load exception", e);
            result = false;
        }

        return result;
    }

    private boolean loadConsumeQueue() {
        File dirLogic = new File(StorePathConfigHelper.getStorePathConsumeQueue(this.messageStoreConfig.getStorePathRootDir()));
        File[] fileTopicList = dirLogic.listFiles();
        if (fileTopicList != null) {

            for (File fileTopic : fileTopicList) {
                String topic = fileTopic.getName();

                File[] fileQueueIdList = fileTopic.listFiles();
                if (fileQueueIdList != null) {
                    for (File fileQueueId : fileQueueIdList) {
                        int queueId;
                        try {
                            queueId = Integer.parseInt(fileQueueId.getName());
                        } catch (NumberFormatException e) {
                            continue;
                        }
                        ConsumeQueue logic = new ConsumeQueue(
                                topic,
                                queueId,
                                StorePathConfigHelper.getStorePathConsumeQueue(this.messageStoreConfig.getStorePathRootDir()),
                                this.getMessageStoreConfig().getMappedFileSizeConsumeQueue(),
                                this);
                        this.putConsumeQueue(topic, queueId, logic);
                        if (!logic.load()) {
                            return false;
                        }
                    }
                }
            }
        }

        log.info("load logics queue all over, OK");

        return true;
    }

    private void putConsumeQueue(final String topic, final int queueId, final ConsumeQueue consumeQueue) {
        ConcurrentMap<Integer/* queueId */, ConsumeQueue> map = this.consumeQueueTable.get(topic);
        if (null == map) {
            map = new ConcurrentHashMap<Integer/* queueId */, ConsumeQueue>();
            map.put(queueId, consumeQueue);
            this.consumeQueueTable.put(topic, map);
        } else {
            map.put(queueId, consumeQueue);
        }
    }

    private PutMessageResult waitForPutResult(CompletableFuture<PutMessageResult> putMessageResultFuture) {
        try {
            int putMessageTimeout = this.messageStoreConfig.getSyncFlushTimeout();
            return putMessageResultFuture.get(putMessageTimeout, TimeUnit.MILLISECONDS);
        } catch (ExecutionException | InterruptedException e) {
            return new PutMessageResult(PutMessageStatus.UNKNOWN_ERROR);
        } catch (TimeoutException e) {
            log.error("usually it will never timeout, putMessageTimeout is much bigger than slaveTimeout and "
                    + "flushTimeout so the result can be got anyway, but in some situations timeout will happen like full gc "
                    + "process hangs or other unexpected situations.");
            return new PutMessageResult(PutMessageStatus.UNKNOWN_ERROR);
        }
    }

    public CompletableFuture<PutMessageResult> asyncPutMessage(MessageExt msg) {
        PutMessageStatus checkStoreStatus = this.checkStoreStatus();
        if (checkStoreStatus != PutMessageStatus.PUT_OK) {
            return CompletableFuture.completedFuture(new PutMessageResult(checkStoreStatus));
        }

        PutMessageStatus msgCheckStatus = this.checkMessage(msg);
        if (msgCheckStatus == PutMessageStatus.MESSAGE_ILLEGAL) {
            return CompletableFuture.completedFuture(new PutMessageResult(msgCheckStatus));
        }

        long beginTime = this.getSystemClock().now();
        CompletableFuture<PutMessageResult> putResultFuture = this.commitLog.asyncPutMsg(msg);

        long elapsedTime = this.getSystemClock().now() - beginTime;
        if (elapsedTime > 500) {
            log.warn("putMessage not in lock elapsed time(ms)={}, bodyLength={}", elapsedTime, msg.getBody().length);
        }
        return putResultFuture;
    }
    private PutMessageStatus checkStoreStatus() {
        if (this.shutdown) {
            log.warn("message store has shutdown, so putMessage is forbidden");
            return PutMessageStatus.SERVICE_NOT_AVAILABLE;
        }

        if (BrokerRole.SLAVE == this.messageStoreConfig.getBrokerRole()) {
            return PutMessageStatus.SERVICE_NOT_AVAILABLE;
        }


        if (this.isOSPageCacheBusy()) {
            return PutMessageStatus.OS_PAGECACHE_BUSY;
        }
        return PutMessageStatus.PUT_OK;
    }

    public boolean isOSPageCacheBusy() {
        long begin = this.getCommitLog().getBeginTimeInLock();
        long diff = this.systemClock.now() - begin;

        return diff < 10000000
                && diff > this.messageStoreConfig.getOsPageCacheBusyTimeOutMills();
    }

    private PutMessageStatus checkMessage(MessageExt msg) {
        if (msg.getTopic().length() > Byte.MAX_VALUE) {
            log.warn("putMessage message topic length too long " + msg.getTopic().length());
            return PutMessageStatus.MESSAGE_ILLEGAL;
        }

        if (msg.getPropertiesString() != null && msg.getPropertiesString().length() > Short.MAX_VALUE) {
            log.warn("putMessage message properties length too long " + msg.getPropertiesString().length());
            return PutMessageStatus.MESSAGE_ILLEGAL;
        }
        return PutMessageStatus.PUT_OK;
    }

    public void shutdown() {
        if (!this.shutdown) {
            this.shutdown = true;

            this.scheduledExecutorService.shutdown();
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                log.error("shutdown Exception, ", e);
            }

            this.indexService.shutdown();
            this.commitLog.shutdown();
            this.reputMessageService.shutdown();
            this.flushConsumeQueueService.shutdown();
        }
        if (lockFile != null && lock != null) {
            try {
                lock.release();
                lockFile.close();
            } catch (IOException e) {
            }
        }
    }

    public String getStorePathPhysic() {
        return DefaultMessageStore.this.getMessageStoreConfig().getStorePathCommitLog();
    }

    public String getStorePathLogic() {
        return StorePathConfigHelper.getStorePathConsumeQueue(this.messageStoreConfig.getStorePathRootDir());
    }

    public SystemClock getSystemClock() {
        return systemClock;
    }

    public void putMessagePositionInfo(DispatchRequest request) {
        ConsumeQueue cq = this.findConsumeQueue(request.getTopic(), request.getQueueId());
        cq.putMessage(request);
    }

    private ConsumeQueue findConsumeQueue(String topic, int queueId) {
        ConcurrentMap<Integer, ConsumeQueue> map = consumeQueueTable.get(topic);
        if (null == map) {
            ConcurrentMap<Integer, ConsumeQueue> newMap = new ConcurrentHashMap<Integer, ConsumeQueue>(128);
            ConcurrentMap<Integer, ConsumeQueue> oldMap = consumeQueueTable.putIfAbsent(topic, newMap);
            if (oldMap != null) {
                map = oldMap;
            } else {
                map = newMap;
            }
        }

        ConsumeQueue logic = map.get(queueId);
        if (null == logic) {
            ConsumeQueue newLogic = new ConsumeQueue(
                    topic,
                    queueId,
                    StorePathConfigHelper.getStorePathConsumeQueue(this.messageStoreConfig.getStorePathRootDir()),
                    this.getMessageStoreConfig().getMappedFileSizeConsumeQueue(),
                    this);
            ConsumeQueue oldLogic = map.putIfAbsent(queueId, newLogic);
            if (oldLogic != null) {
                logic = oldLogic;
            } else {
                logic = newLogic;
            }
        }

        return logic;
    }

    public long getConfirmOffset() {
        return this.commitLog.getConfirmOffset();
    }


    public SelectMappedBufferResult selectConsumeQueue(String topic,Integer queueId,long consumeOffset){
        try{
            SelectMappedBufferResult select = consumeQueueTable.get(topic).get(queueId).getIndexBuffer(consumeOffset);
            return select;
        }catch (Exception e){
            return null;
        }
    }
    public SelectMappedBufferResult selectCommitLog(String topic,Integer queueId,long consumeOffset){
        //先查consume queue
        SelectMappedBufferResult select = selectConsumeQueue(topic, queueId, consumeOffset);
        ByteBuffer byteBuffer = select.getByteBuffer();
        long commitLogOffset = byteBuffer.getLong();
        int msgSize = byteBuffer.getInt();
        long tag = byteBuffer.getLong();
        //根据commit log offset查commit log
        return select(commitLogOffset,msgSize);
    }

    public SelectMappedBufferResult select(long offset,int size) {
        return this.commitLog.select(offset, size);
    }

    public SelectMappedBufferResult select(int offset) {
        SelectMappedBufferResult sbr = this.commitLog.select(offset, 4);
        if (null != sbr) {
            try {
                // 1 TOTALSIZE
                int size = sbr.getByteBuffer().getInt();
                return this.commitLog.select(offset, size);
            } finally {
                sbr.release();
            }
        }
        return null;
    }

    class CommitLogDispatcherBuildConsumeQueue implements CommitLogDispatcher {

        @Override
        public void dispatch(DispatchRequest request) {
            DefaultMessageStore.this.putMessagePositionInfo(request);
        }
    }




    class CommitLogDispatcherBuildIndex implements CommitLogDispatcher {

        @Override
        public void dispatch(DispatchRequest request) {
            if (DefaultMessageStore.this.messageStoreConfig.isMessageIndexEnable()) {
                DefaultMessageStore.this.indexService.buildIndex(request);
            }
        }
    }

    class ReputMessageService extends ServiceThread {

        private long reputFromOffset;
        public ReputMessageService(long reputFromOffset){
            this.reputFromOffset = reputFromOffset;
        }

        @Override
        public String getServiceName() {
            return ReputMessageService.class.getSimpleName();
        }

        public long behind() {
            return DefaultMessageStore.this.commitLog.getMaxOffset() - this.reputFromOffset;
        }

        private boolean isCommitLogAvailable() {
            return this.reputFromOffset < DefaultMessageStore.this.commitLog.getMaxOffset();
        }

        @Override
        public void run() {
            DefaultMessageStore.log.info(this.getServiceName() + " service started");

            while (!this.isStopped()) {
                try {
                    Thread.sleep(1);
                    this.doReput();
                } catch (Exception e) {
                    DefaultMessageStore.log.warn(this.getServiceName() + " service has exception. ", e);
                }
            }

            DefaultMessageStore.log.info(this.getServiceName() + " service end");
        }

        private void doReput(){
            if (this.reputFromOffset < DefaultMessageStore.this.commitLog.getMinOffset()) {
                log.warn("The reputFromOffset={} is smaller than minPyOffset={}, this usually indicate that the dispatch behind too much and the commitlog has expired.",
                        this.reputFromOffset, DefaultMessageStore.this.commitLog.getMinOffset());
                this.reputFromOffset = DefaultMessageStore.this.commitLog.getMinOffset();
            }

            if(this.isCommitLogAvailable()){
                SelectMappedBufferResult select = DefaultMessageStore.this.commitLog.select(reputFromOffset);
                if(null!=select){
                    log.info("reputFromOffset="+reputFromOffset);
                    this.reputFromOffset = select.getStartOffset()+select.getSize();

                    DispatchRequest request = DefaultMessageStore.this.commitLog.populateDispatchRequest(select.getByteBuffer());

                    DefaultMessageStore.this.doDispatch(request);
                }
            }
        }


        public void setReputFromOffset(long reputFromOffset) {
            this.reputFromOffset = reputFromOffset;
        }
    }

    private void doDispatch(DispatchRequest request) {
        for (CommitLogDispatcher dispatcher : dispatcherList) {
            dispatcher.dispatch(request);
        }
    }


    class FlushConsumeQueueService extends  ServiceThread{
        private static final int RETRY_TIMES_OVER = 3;

        @Override
        public String getServiceName() {
            return FlushConsumeQueueService.class.getSimpleName();
        }

        @Override
        public void run() {
            DefaultMessageStore.log.info(this.getServiceName() + " service started");

            while(!this.isStopped()){
                try {
                    int interval = DefaultMessageStore.this.getMessageStoreConfig().getFlushIntervalConsumeQueue();
                    this.waitForRunning(interval);
                    this.doFlush(1);
                } catch (Exception e) {
                    DefaultMessageStore.log.warn(this.getServiceName() + " service has exception. ", e);
                }
            }

            DefaultMessageStore.log.info(this.getServiceName() + " service end");
        }

        private void doFlush(int retryTimes) {
            int flushConsumeQueueLeastPages = DefaultMessageStore.this.getMessageStoreConfig().getFlushConsumeQueueLeastPages();

            if (retryTimes == RETRY_TIMES_OVER) {
                flushConsumeQueueLeastPages = 0;
            }
            ConcurrentMap<String, ConcurrentMap<Integer, ConsumeQueue>> tables = DefaultMessageStore.this.consumeQueueTable;

            for (ConcurrentMap<Integer, ConsumeQueue> maps : tables.values()) {
                for (ConsumeQueue cq : maps.values()) {
                    boolean result = false;
                    for(int i=0;i<retryTimes&&!result;i++){
                        result = cq.flush(flushConsumeQueueLeastPages);
                    }
                }
            }
        }

        @Override
        public long getJointime() {
            return 1000 * 60;
        }
    }
}
